package cn.zhang.violet.report


import cn.zhang.violet.bean.RptArea
import cn.zhang.violet.config.ConfigHandler
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{SQLContext, SaveMode}

object AppReportCore {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf()
    sparkConf.setAppName("地域分布")
    sparkConf.setMaster("local[*]")
    sparkConf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")

    val sc = new SparkContext(sparkConf)

    val sQLContext = new SQLContext(sc)
    //读取数据
    val dataFrame = sQLContext.read.parquet(ConfigHandler.parquetFilePath)

    //需要导入隐式转换
    import sQLContext.implicits._


   dataFrame.map(str => {
      val privName = str.getAs[String]("provincename")
      val cityName = str.getAs[String]("cityname")

     // 提取个指标相关的字段
     val reqMode = str.getAs[Int]("requestmode")
     val proNode = str.getAs[Int]("processnode")
     val effTive = str.getAs[Int]("iseffective")
     val billing = str.getAs[Int]("isbilling")
     val isbid = str.getAs[Int]("isbid")
     val iswin = str.getAs[Int]("iswin")
     val adOrderId = str.getAs[Int]("adorderid")

     val rawReq = if(reqMode == 1 && proNode >= 1) 1 else 0
     val effReq = if(reqMode == 1 && proNode >= 2) 1 else 0
     val adReq =if(reqMode == 1 && proNode >= 3) 1 else 0

     val adRtbReq = if(effTive == 1 && billing == 1 && isbid == 1 && adOrderId != 0) 1 else 0
     val adSuccRtbAndCostAndConsumption = if(effTive == 1 && billing == 1 && iswin == 1){

       val winPrice = str.getAs[Double]("winprice")
       val adPayment = str.getAs[Double]("adpayment")
       (1,adPayment /1000,winPrice /1000)

     } else (0,0d,0d)
     val adShow = if(reqMode == 2 && effTive == 1) 1 else 0
     val adClick = if(reqMode == 3 && effTive == 1) 1 else 0


     //组成一个对偶元祖
     ((privName,cityName),List(rawReq, effReq, adReq, adRtbReq, adSuccRtbAndCostAndConsumption._1,
       adShow, adClick, adSuccRtbAndCostAndConsumption._2, adSuccRtbAndCostAndConsumption._3))
   }).reduceByKey((a,b) =>a.zip(b).map(t => t._1 + t._2))
      .map(tp => RptArea (
        tp._1._1,
        tp._1._2,
        tp._2(0).toInt,
        tp._2(1).toInt,
        tp._2(2).toInt,
        tp._2(3).toInt,
        tp._2(4).toInt,
        tp._2(5).toInt,
        tp._2(6).toInt,
        tp._2(7),
        tp._2(8)
      )).toDF().write.mode(SaveMode.Overwrite)
     .jdbc(ConfigHandler.url, "violet_prodata_joinnum", ConfigHandler.props)




    sc.stop()
  }
}
